package cn.com.cennavi.localModel;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairInputDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

public class Test01 {
	public static void main(String[] args) {
		SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount");
		JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(1));
		JavaReceiverInputDStream<String> lines = jssc.socketTextStream("localhost", 9999);
		// JavaPairInputDStream<Object, Object> lines =
		// jssc.fileStream("/APP/lanePrograme/laneData/didi/0725.dat");
		JavaDStream<String> ss=lines.map(new Function<String, String>() {

			@Override
			public String call(String arg0) throws Exception {
				// TODO Auto-generated method stubste
//				System.out.println(arg0);
				return arg0;
			}
		});
		ss.print();
		// // Count each word in each batch
		// JavaPairDStream<String, Integer> pairs = words.mapToPair(new
		// PairFunction<String, String, Integer>() {
		// public Tuple2<String, Integer> call(String s) {
		// return new Tuple2<String, Integer>(s, 1);
		// }
		// });
		// JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey(new
		// Function2<Integer, Integer, Integer>() {
		// public Integer call(Integer i1, Integer i2) {
		// return i1 + i2;
		// }
		// });
		//
		// // Print the first ten elements of each RDD generated in this DStream
		// to
		// // the console
		// wordCounts.print();

		jssc.start(); // Start the computation
//		jssc.awaitTermination();
	}
}
